package com.nl.networkflow;


import com.nl.bean.input.ApacheLogEvent;
import com.nl.bean.output.WindowAggResult;
import com.nl.utils.WindowAggResultOut;
import com.nl.utils.aggfunctions.CountAgg;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author shihb
 * @date 2019/12/19 18:36
 * 热门页面统计：每10秒输出10分钟内访问最多的url
 */
public class NetWorkFlow {

  public static void main(String[] args) throws Exception {
    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // 设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    //2.source
    DataStreamSource<String> source = env
        .readTextFile(NetWorkFlow.class.getClassLoader().getResource("apache.log").getPath());
    //3.transform算子
    SingleOutputStreamOperator<String> dataStream = source
        .map(s->{
          String[] arr = s.split(" ");
          // 定义事件转换
          SimpleDateFormat sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
          long timestamps = sdf.parse(arr[3]).getTime();
          return new ApacheLogEvent(arr[0].trim(), timestamps, arr[5].trim(), arr[6].trim());
        })
        // 提取事件时间
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<ApacheLogEvent>(Time.seconds(1)){
              @Override
              public long extractTimestamp(ApacheLogEvent element) {
                return element.getTimeStamp();
              }
            }
        )
        .filter(apacheLogEvent->apacheLogEvent.getUrl().contains(".html"))
        // 根据url分组
        .keyBy(apacheLogEvent->apacheLogEvent.getUrl())
        // 设置滑动窗口
        .timeWindow(Time.minutes(10), Time.seconds(10))
        // 允许60秒的迟到更新
        .allowedLateness(Time.seconds(60))
        // 预聚合函数
        .aggregate(new CountAgg<ApacheLogEvent>(), new WindowAggResultOut<String>())
        // 根据窗口再分组
        .keyBy(windowAggResult -> windowAggResult.getWindowEnd())
        // 排序输出
        .process(new TopnDataFlow(3));

    //4.sink
    dataStream.print();

    //5.执行
    env.execute("network flow job");


  }

}
class TopnDataFlow extends KeyedProcessFunction<Long, WindowAggResult,String> {

  private int topSize;
  private ListState<WindowAggResult> urlState;

  public TopnDataFlow(int i) {
    topSize=i;
  }

  @Override
  public void open(Configuration parameters) throws Exception {
    urlState=getRuntimeContext().getListState(new ListStateDescriptor<WindowAggResult>("dataFlowState",
        WindowAggResult.class));
  }

  @Override
  public void processElement(WindowAggResult value, Context ctx, Collector<String> out)
      throws Exception {
      urlState.add(value);
      ctx.timerService().registerEventTimeTimer(value.getWindowEnd()+1);
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
    List<WindowAggResult> allUrlState = new ArrayList<WindowAggResult>();
    // 拿到该窗口的所有状态
    for(WindowAggResult ucv:urlState.get()){
      allUrlState.add(ucv);
    };
    urlState.clear();
    // 排序
    allUrlState.sort((o1,o2)->(int)(o2.getCount()-o1.getCount()));
    // 输出
    StringBuilder sb = new StringBuilder();
    sb.append("时间：").append(new Timestamp(timestamp-1)).append("\n");

    for (int i = 0; i <Math.min(topSize,allUrlState.size()) ; i++) {
      WindowAggResult ucv = allUrlState.get(i);
      sb.append("No").append(i+1).append(":")
          .append("url=").append(ucv.getKey()).append(" ")
          .append("访问量=").append(ucv.getCount()).append("\n");

    }
    sb.append("======================");
    out.collect(sb.toString());


  }
}


